package com.shujia.flink.kafka

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

import scala.io.Source

object Demo1Peoduce {

  def main(args: Array[String]): Unit = {


    /**
      * 创建生产者
      *
      */

    val properties = new Properties
    //kafka broker 地址列表
    properties.setProperty("bootstrap.servers", "master:9092,node1:9092,node2:9092")
    //key和value序列化的方式
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")


    val producer = new KafkaProducer[String, String](properties)


    /**
      * 生产数据
      *
      */


    val students: List[String] = Source.fromFile("flink/data/students.txt").getLines().toList


    for (student <- students) {


      /**
        * topic 可以不存在，在生产数据的时候会自动创建，自动创建一个分区数为1  副本数为1 的topic
        *
        */

      //将同一个班级的数据存到同一个分区中
      //好处：  后续按照班级进行分组，效率高
      // 缺点：  可能会导致数据倾斜
      //kafka-topics.sh --create --zookeeper master:2181 --replication-factor 3 --partitions 3 --topic student1

      val clazz: String = student.split(",")(4)

      //hash   计算班级属于哪个分区
      val partition: Int = Math.abs(clazz.hashCode) % 3


      val record = new ProducerRecord[String, String]("student1", partition, null, student)



      //发送数据
      producer.send(record)

      //刷新到kafka就请你
      producer.flush()


    }



    //关闭链接
    producer.close()


  }

}
